/**
 * Software License, Version 1.0
 * 
 * Copyright 2008 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgment:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgment may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or Granules, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called Granules, 
 * nor may Indiana University or Community Grids Lab or Granules appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 * LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 * WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 * NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 * INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 * INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 * "VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 * LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 * ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 * GENERATED USING SOFTWARE.
 */
package cgl.granules.samples.grep;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.Properties;
import java.util.StringTokenizer;

import cgl.granules.dataset.DatasetCollection;
import cgl.granules.dataset.DatasetException;
import cgl.granules.dataset.DatasetFactory;
import cgl.granules.dataset.FileAccess;
import cgl.granules.exception.CommunicationsException;
import cgl.granules.exception.DeploymentException;
import cgl.granules.exception.MarshallingException;
import cgl.granules.operation.MapReduceException;
import cgl.granules.operation.ProgressTracker;
import cgl.granules.operation.InstanceDeployer;

/**
 * The deployer for the GREP information retrieval example, which allows one to
 * perform a regular expressions query on given set of files over a distributed
 * set of resources.
 * 
 * @author Shrideep Pallickara
 * 
 */
public class GrepCountDeployer extends InstanceDeployer {
    private ProgressTracker progressTracker;


    /**
     * Initializes the deployer.
     * 
     * @param streamingProperties
     *            The properties for streaming
     * @throws CommunicationsException
     *             If there are network problems during communications
     * @throws IOException
     *             If there are IO problems
     * @throws MarshallingException
     *             If there are marshalling erros
     * @throws DeploymentException
     *             If there are deployment related problems
     */
    public GrepCountDeployer(Properties streamingProperties)
        throws CommunicationsException, IOException, MarshallingException,
        DeploymentException {
        initialize(streamingProperties);
    }


    /**
     * This method distributes a set of files over the map instances. The method
     * requires the file names to be numbered from 1 .... Thus, if the files are
     * file1.txt file2.txt etc the filename base that needs to be specified
     * would be /location/Of/Files/file
     * 
     * 
     * @param reduce
     *            The Reducer for this example.
     * @param mapNum
     *            The map instance number
     * @param filenameBase
     *            The directory in which the files are stored.
     * @param numOfMaps
     *            The total number of maps being deployed
     * @param numOfFiles
     *            The total number of files that need to be processed
     * @return The Initialized <code>GrepCountMap</code> instance.
     * @throws DatasetException
     *             If there are problems initializing/using the datasets
     * @throws MapReduceException
     *             If there are problems related to Map/Reduce roles within the
     *             application.
     */
    private GrepCountMap initializeGrepCountMap(GrepCountReduce reduce,
                                                int mapNum,
                                                String filenameBase,
                                                int numOfMaps, int numOfFiles)
        throws DatasetException, MapReduceException {
        GrepCountMap map = new GrepCountMap();

        Properties props = new Properties();
        String value = "(" + mapNum + ")";
        String property = "Worker Number";
        props.put(property, value);
        map.setProcessingDirectives(props);

        DatasetFactory datasetFactory = DatasetFactory.getInstance();
        FileAccess fileAccess =
            datasetFactory.createFileDataset("FileDataset-ID-GrepCount",
                                             "Grep Count example");

        for (int i = 0; i < numOfFiles; i++) {
            if (i % numOfMaps == mapNum) {
                String filename = filenameBase + i + ".txt";
                System.out.println(filename + "-> assigned to MAP [" + mapNum
                    + "]");
                fileAccess.addFile(filename, "rw");
            }/* the modulo-check */
        }

        DatasetCollection datasetCollection = map.getDatasetCollection();
        datasetCollection.addDataset(fileAccess);

        map.addReduce(reduce);

        return map;
    }


    /**
     * Prepares a set of Map/Reduce instances for the Grep example and deploys
     * it on the set of available resources.
     * 
     * @param numOfMaps
     *            The total number of maps being deployed
     * @param filenameBase
     *            The directory in which the files are stored.
     * @param numOfFiles
     *            The total number of files that need to be processed
     * @throws DatasetException
     *             If there are problems initializing the datasets
     * @throws MapReduceException
     *             If there is a MapReduceExcecution because of problems when
     *             setting up the roles.
     * @throws IOException
     *             If there are IO problems.
     * @throws CommunicationsException
     *             If there are network problems during communications
     * @throws DeploymentException
     *             If there are deployment related problems
     * @throws MarshallingException
     *             If there are marshalling problems.
     * @throws InterruptedException
     *             If there are problems sleeping
     */
    public void prepareInstancesAndDeploy(int numOfMaps, String filenameBase,
                                          int numOfFiles)
        throws MapReduceException, CommunicationsException,
        DeploymentException, MarshallingException, DatasetException,
        InterruptedException {

        GrepCountMap maps[] = new GrepCountMap[numOfMaps];

        GrepCountReduce reduce = new GrepCountReduce();
        reduce.setAsExecuteWhenDataAvailable();
        GrepCountReduce reducers[] = new GrepCountReduce[1];
        reducers[0] = reduce;

        for (int i = 0; i < numOfMaps; i++) {
            maps[i] =
                initializeGrepCountMap(reduce, i, filenameBase, numOfMaps,
                                       numOfFiles);
        }

        resetTrackers();

        progressTracker = deployOperations(reducers);
        String jobId = progressTracker.getIdentifier();

        Thread.sleep(1000);

        deployOperations(maps, jobId);
    }


    /**
     * @param args
     */
    public static void main(String[] args) {
        String module = "GrepCountDeployer.main()->";
        if (args.length < 2) {
            System.out.println("Usage: java cgl.granules.samples.grep."
                + "GrepCountDeployer <hostname> <port>");
            System.exit(0);
        }
        String hostname = args[0];
        String portnum = args[1];
        int numOfMaps = 2;
        int numOfFiles = 2;
        String filenameBase = "C:/Granules/Datasets/Files/file";

        Properties streamingProperties = new Properties();
        streamingProperties.put("hostname", hostname);
        streamingProperties.put("portnum", portnum);

        String helpInfo =
            "\n\nThis is the GREP test program that uses Granules."
                + "The current list of commands include: \n"
                + "submit <filename_base> <num_of_maps> <num_of_resources> \n"
                + "status        : Lists the status of the all Jobs\n"
                + "h             : List the set of commands \n\n";
        try {
            GrepCountDeployer deployer =
                new GrepCountDeployer(streamingProperties);
            StringTokenizer tok;

            InputStreamReader isr = new InputStreamReader(System.in);
            BufferedReader commandLine = new BufferedReader(isr);

            System.out.println(helpInfo);

            while (true) {
                String s = commandLine.readLine();
                if (s == null) {
                    System.out.println(module + "String is null!!!");
                    break;
                }

                if (s.startsWith("submit")) {
                    System.out.println(module
                        + "Processing the [submit] command");

                    tok = new StringTokenizer(s, " ");
                    int numOfTokens = tok.countTokens();
                    tok.nextToken();/* skip the command */

                    if (numOfTokens > 1) {
                        filenameBase = tok.nextToken();
                    }

                    if (numOfTokens > 2) {
                        numOfMaps = Integer.parseInt(tok.nextToken());
                    }

                    if (numOfTokens > 3) {
                        numOfFiles = Integer.parseInt(tok.nextToken());
                    }

                    String params =
                        "\nFilename base      : " + filenameBase
                            + "\nNumber of Maps     : " + numOfMaps
                            + "\nNumber of Files    :" + numOfFiles;
                    System.out.println(params);

                    deployer.prepareInstancesAndDeploy(numOfMaps, filenameBase,
                                                       numOfFiles);

                    continue;
                }

                if (s.startsWith("status")) {
                    System.out.println(deployer.getStatus());
                    continue;
                }
            }

        } catch (Exception e) {
            System.out.println(e.getMessage());
            e.printStackTrace();
        }
    }
}
